﻿using System.Collections.Generic;
using Apache.Cassandra;
using Thrift.Protocol;
using Thrift.Transport;

namespace SKBKontur.Cassandra.Proxy
{
    public class Storage : IStorage
    {
        private readonly Apache.Cassandra.Cassandra.Client client;
        private readonly TTransport transport;

        internal Storage(TTransport transport, string keySpace)
        {
            this.transport = transport;
            var protocol = new TBinaryProtocol(transport);
            client = new Apache.Cassandra.Cassandra.Client(protocol);
            client.set_keyspace(keySpace);
        }

        #region IStorage Members

        public void Insert(byte[] key, ColumnParent columnParent, ConsistencyLevel consistencyLevel,
                           params Column[] columns)
        {
            foreach (Column column in columns)
                client.insert(key, columnParent, column, consistencyLevel);
        }

        public Dictionary<byte[], List<ColumnOrSuperColumn>> MultigetSlice(IEnumerable<byte[]> keys,
                                                                           ColumnParent columnParent,
                                                                           SlicePredicate slicePredicate,
                                                                           ConsistencyLevel consistencyLevel)
        {
            return client.multiget_slice(new List<byte[]>(keys), columnParent, slicePredicate, consistencyLevel);
        }

        public List<ColumnOrSuperColumn> GetSlice(byte[] key, ColumnParent columnParent, SlicePredicate slicePredicate,
                                                  ConsistencyLevel consistencyLevel)
        {
            return client.get_slice(key, columnParent, slicePredicate, consistencyLevel);
        }

        public List<KeySlice> GetIndexedSlice(ColumnParent columnParent, IndexClause indexClause, SlicePredicate column_predicate, ConsistencyLevel consistencyLevel)
        {
            return client.get_indexed_slices(columnParent, indexClause, column_predicate, ConsistencyLevel.ALL);
        }

        public void Dispose()
        {
            transport.Close();
        }

        #endregion
    }
}